package com.wzz.shop.realtime.etl.`trait`

import com.wzz.shop.realtime.etl.utils.KafkaProps
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._

/**
 *
 * @author ${author}
 * @date 2021-09-14 19:41
 *
 */
abstract class  MQBaseETL(env:StreamExecutionEnvironment) extends BaseETL[String] {
  override def getKafkaDataStream(topic: String): DataStream[String] = {

    val kafkaProducer: FlinkKafkaConsumer011[String] = new FlinkKafkaConsumer011[String](
    topic,
    new SimpleStringSchema(),
    KafkaProps.getKafkaProperties()
    )
    val logDataStream: DataStream[String] = env.addSource(kafkaProducer)

    //返回消费到的数据
    logDataStream
  }
}
